package com.wangyh8216.demo.flink;

// Flink类型系统相关导入
import org.apache.flink.api.common.typeinfo.Types;
// 批处理执行环境
import org.apache.flink.api.java.ExecutionEnvironment;
// Flink批处理操作符导入
import org.apache.flink.api.java.operators.AggregateOperator;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.operators.FlatMapOperator;
import org.apache.flink.api.java.operators.UnsortedGrouping;
// Flink元组类型
import org.apache.flink.api.java.tuple.Tuple2;
// Flink数据收集器
import org.apache.flink.util.Collector;

/**
 * 批处理单词计数示例程序
 * 演示Flink批处理API的基本使用方法
 * 功能：统计文本文件中每个单词的出现次数
 * 
 * @author wangyh8216
 * @version 1.0
 * @since 2024
 */
public class BatchWordCount {

    /**
     * 主方法 - 批处理单词计数程序的入口点
     * 执行流程：
     * 1. 创建执行环境
     * 2. 读取文本文件数据
     * 3. 分词并转换为(单词, 1)元组
     * 4. 按单词分组
     * 5. 统计每个单词的出现次数
     * 6. 输出结果
     * 
     * @param args 命令行参数（未使用）
     */
    public static void main(String[] args) {
        // 1、创建批处理执行环境
        // ExecutionEnvironment是Flink批处理作业的入口点
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

        // 2、从文件读取数据源
        // 读取src/main/resources/words.txt文件中的文本数据
        // DataSource<String>表示字符串类型的数据源
        DataSource<String> stringDataSource = env.readTextFile("src/main/resources/words.txt");

        // 3、数据转换：将每行文本分词并转换为(单词, 1)二元组
        // flatMap操作将每行文本拆分为多个单词，并为每个单词生成计数1
        FlatMapOperator<String, Tuple2<String, Long>> wordTuple = stringDataSource.flatMap((String line, Collector<Tuple2<String, Long>> out) -> {
            // 使用空格分隔符将一行文本拆分为单词数组
            String[] words = line.split(" ");
            // 遍历每个单词，输出(单词, 1)元组
            for (String word : words) {
                // 收集器输出元组，格式为(单词, 计数1)
                out.collect(Tuple2.of(word, 1L));
            }
        // 显式指定返回类型，避免Java泛型类型擦除问题
        // Types.TUPLE定义元组类型，Types.STRING表示字符串类型，Types.LONG表示长整型
        }).returns(Types.TUPLE(Types.STRING, Types.LONG));

        // 4、数据分组：按照单词字段进行分组
        // groupBy(0)表示按元组的第一个字段（即单词）进行分组
        // UnsortedGrouping表示分组后的数据集合
        UnsortedGrouping<Tuple2<String, Long>> wordGroup = wordTuple.groupBy(0);

        // 5、聚合统计：在分组内对计数字段进行求和
        // sum(1)表示对元组的第二个字段（即计数）进行求和操作
        // AggregateOperator表示聚合操作的结果
        AggregateOperator<Tuple2<String, Long>> sum = wordGroup.sum(1);

        try {
            // 输出聚合结果到控制台
            // print()方法会触发作业执行并显示结果
            sum.print();
        } catch (Exception e) {
            // 异常处理：打印异常堆栈信息
            e.printStackTrace();
        }
    }
}
